"""Test service helpers."""

import asyncio
from collections.abc import Callable, Iterable
from copy import deepcopy
import dataclasses
import io
import threading
from typing import Any
from unittest.mock import AsyncMock, Mock, patch

import pytest
from pytest_unordered import unordered
import voluptuous as vol

# To prevent circular import when running just this file
from homeassistant import exceptions
from homeassistant.auth.permissions import PolicyPermissions
import homeassistant.components  # noqa: F401
from homeassistant.components.group import DOMAIN as DOMAIN_GROUP, Group
from homeassistant.components.input_button import DOMAIN as DOMAIN_INPUT_BUTTON
from homeassistant.components.logger import DOMAIN as DOMAIN_LOGGER
from homeassistant.components.shell_command import DOMAIN as DOMAIN_SHELL_COMMAND
from homeassistant.components.system_health import DOMAIN as DOMAIN_SYSTEM_HEALTH
from homeassistant.const import (
    ATTR_ENTITY_ID,
    ENTITY_MATCH_ALL,
    ENTITY_MATCH_NONE,
    STATE_OFF,
    STATE_ON,
    EntityCategory,
)
from homeassistant.core import (
    Context,
    HassJob,
    HomeAssistant,
    ServiceCall,
    ServiceResponse,
    SupportsResponse,
    callback,
)
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import (
    area_registry as ar,
    config_validation as cv,
    device_registry as dr,
    entity_registry as er,
    service,
)
from homeassistant.loader import (
    Integration,
    async_get_integration,
    async_get_integrations,
)
from homeassistant.setup import async_setup_component
from homeassistant.util.yaml.loader import JSON_TYPE, parse_yaml

from tests.common import (
    MockEntity,
    MockEntityPlatform,
    MockModule,
    MockUser,
    RegistryEntryWithDefaults,
    async_mock_service,
    mock_area_registry,
    mock_device_registry,
    mock_integration,
    mock_registry,
)

SUPPORT_A = 1
SUPPORT_B = 2
SUPPORT_C = 4


@pytest.fixture
def mock_handle_entity_call():
    """Mock service platform call."""
    with patch(
        "homeassistant.helpers.service._handle_entity_call",
        return_value=None,
    ) as mock_call:
        yield mock_call


@pytest.fixture
def mock_entities(hass: HomeAssistant) -> dict[str, MockEntity]:
    """Return mock entities in an ordered dict."""
    kitchen = MockEntity(
        entity_id="light.kitchen",
        available=True,
        should_poll=False,
        supported_features=SUPPORT_A,
        device_class=None,
    )
    living_room = MockEntity(
        entity_id="light.living_room",
        available=True,
        should_poll=False,
        supported_features=SUPPORT_B,
        device_class="class_a",
    )
    bedroom = MockEntity(
        entity_id="light.bedroom",
        available=True,
        should_poll=False,
        supported_features=(SUPPORT_A | SUPPORT_B),
        device_class="class_b",
    )
    bathroom = MockEntity(
        entity_id="light.bathroom",
        available=True,
        should_poll=False,
        supported_features=(SUPPORT_B | SUPPORT_C),
        device_class="class_c",
    )
    entities = {}
    entities[kitchen.entity_id] = kitchen
    entities[living_room.entity_id] = living_room
    entities[bedroom.entity_id] = bedroom
    entities[bathroom.entity_id] = bathroom
    for entity in entities.values():
        entity.hass = hass
    return entities


@pytest.fixture
def floor_area_mock(hass: HomeAssistant) -> None:
    """Mock including floor and area info."""
    hass.states.async_set("light.Bowl", STATE_ON)
    hass.states.async_set("light.Ceiling", STATE_OFF)
    hass.states.async_set("light.Kitchen", STATE_OFF)

    area_in_floor = ar.AreaEntry(
        id="test-area",
        name="Test area",
        aliases={},
        floor_id="test-floor",
        icon=None,
        picture=None,
        temperature_entity_id=None,
        humidity_entity_id=None,
    )
    area_in_floor_a = ar.AreaEntry(
        id="area-a",
        name="Area A",
        aliases={},
        floor_id="floor-a",
        icon=None,
        picture=None,
        temperature_entity_id=None,
        humidity_entity_id=None,
    )
    mock_area_registry(
        hass,
        {
            area_in_floor.id: area_in_floor,
            area_in_floor_a.id: area_in_floor_a,
        },
    )

    device_in_area = dr.DeviceEntry(area_id="test-area")
    device_no_area = dr.DeviceEntry(id="device-no-area-id")
    device_diff_area = dr.DeviceEntry(area_id="diff-area")
    device_area_a = dr.DeviceEntry(id="device-area-a-id", area_id="area-a")

    mock_device_registry(
        hass,
        {
            device_in_area.id: device_in_area,
            device_no_area.id: device_no_area,
            device_diff_area.id: device_diff_area,
            device_area_a.id: device_area_a,
        },
    )

    entity_in_own_area = RegistryEntryWithDefaults(
        entity_id="light.in_own_area",
        unique_id="in-own-area-id",
        platform="test",
        area_id="own-area",
    )
    config_entity_in_own_area = RegistryEntryWithDefaults(
        entity_id="light.config_in_own_area",
        unique_id="config-in-own-area-id",
        platform="test",
        area_id="own-area",
        entity_category=EntityCategory.CONFIG,
    )
    hidden_entity_in_own_area = RegistryEntryWithDefaults(
        entity_id="light.hidden_in_own_area",
        unique_id="hidden-in-own-area-id",
        platform="test",
        area_id="own-area",
        hidden_by=er.RegistryEntryHider.USER,
    )
    entity_in_area = RegistryEntryWithDefaults(
        entity_id="light.in_area",
        unique_id="in-area-id",
        platform="test",
        device_id=device_in_area.id,
    )
    config_entity_in_area = RegistryEntryWithDefaults(
        entity_id="light.config_in_area",
        unique_id="config-in-area-id",
        platform="test",
        device_id=device_in_area.id,
        entity_category=EntityCategory.CONFIG,
    )
    hidden_entity_in_area = RegistryEntryWithDefaults(
        entity_id="light.hidden_in_area",
        unique_id="hidden-in-area-id",
        platform="test",
        device_id=device_in_area.id,
        hidden_by=er.RegistryEntryHider.USER,
    )
    entity_in_other_area = RegistryEntryWithDefaults(
        entity_id="light.in_other_area",
        unique_id="in-area-a-id",
        platform="test",
        device_id=device_in_area.id,
        area_id="other-area",
    )
    entity_assigned_to_area = RegistryEntryWithDefaults(
        entity_id="light.assigned_to_area",
        unique_id="assigned-area-id",
        platform="test",
        device_id=device_in_area.id,
        area_id="test-area",
    )
    entity_no_area = RegistryEntryWithDefaults(
        entity_id="light.no_area",
        unique_id="no-area-id",
        platform="test",
        device_id=device_no_area.id,
    )
    config_entity_no_area = RegistryEntryWithDefaults(
        entity_id="light.config_no_area",
        unique_id="config-no-area-id",
        platform="test",
        device_id=device_no_area.id,
        entity_category=EntityCategory.CONFIG,
    )
    hidden_entity_no_area = RegistryEntryWithDefaults(
        entity_id="light.hidden_no_area",
        unique_id="hidden-no-area-id",
        platform="test",
        device_id=device_no_area.id,
        hidden_by=er.RegistryEntryHider.USER,
    )
    entity_diff_area = RegistryEntryWithDefaults(
        entity_id="light.diff_area",
        unique_id="diff-area-id",
        platform="test",
        device_id=device_diff_area.id,
    )
    entity_in_area_a = RegistryEntryWithDefaults(
        entity_id="light.in_area_a",
        unique_id="in-area-a-id",
        platform="test",
        device_id=device_area_a.id,
        area_id="area-a",
    )
    entity_in_area_b = RegistryEntryWithDefaults(
        entity_id="light.in_area_b",
        unique_id="in-area-b-id",
        platform="test",
        device_id=device_area_a.id,
        area_id="area-b",
    )
    mock_registry(
        hass,
        {
            entity_in_own_area.entity_id: entity_in_own_area,
            config_entity_in_own_area.entity_id: config_entity_in_own_area,
            hidden_entity_in_own_area.entity_id: hidden_entity_in_own_area,
            entity_in_area.entity_id: entity_in_area,
            config_entity_in_area.entity_id: config_entity_in_area,
            hidden_entity_in_area.entity_id: hidden_entity_in_area,
            entity_in_other_area.entity_id: entity_in_other_area,
            entity_assigned_to_area.entity_id: entity_assigned_to_area,
            entity_no_area.entity_id: entity_no_area,
            config_entity_no_area.entity_id: config_entity_no_area,
            hidden_entity_no_area.entity_id: hidden_entity_no_area,
            entity_diff_area.entity_id: entity_diff_area,
            entity_in_area_a.entity_id: entity_in_area_a,
            entity_in_area_b.entity_id: entity_in_area_b,
        },
    )


@pytest.fixture
def label_mock(hass: HomeAssistant) -> None:
    """Mock including label info."""
    hass.states.async_set("light.bowl", STATE_ON)
    hass.states.async_set("light.ceiling", STATE_OFF)
    hass.states.async_set("light.kitchen", STATE_OFF)

    area_with_labels = ar.AreaEntry(
        id="area-with-labels",
        name="Area with labels",
        aliases={},
        floor_id=None,
        icon=None,
        labels={"label_area"},
        picture=None,
        temperature_entity_id=None,
        humidity_entity_id=None,
    )
    area_without_labels = ar.AreaEntry(
        id="area-no-labels",
        name="Area without labels",
        aliases={},
        floor_id=None,
        icon=None,
        labels=set(),
        picture=None,
        temperature_entity_id=None,
        humidity_entity_id=None,
    )
    mock_area_registry(
        hass,
        {
            area_with_labels.id: area_with_labels,
            area_without_labels.id: area_without_labels,
        },
    )

    device_has_label1 = dr.DeviceEntry(labels={"label1"})
    device_has_label2 = dr.DeviceEntry(labels={"label2"})
    device_has_labels = dr.DeviceEntry(
        labels={"label1", "label2"}, area_id=area_with_labels.id
    )
    device_no_labels = dr.DeviceEntry(
        id="device-no-labels", area_id=area_without_labels.id
    )

    mock_device_registry(
        hass,
        {
            device_has_label1.id: device_has_label1,
            device_has_label2.id: device_has_label2,
            device_has_labels.id: device_has_labels,
            device_no_labels.id: device_no_labels,
        },
    )

    entity_with_my_label = RegistryEntryWithDefaults(
        entity_id="light.with_my_label",
        unique_id="with_my_label",
        platform="test",
        labels={"my-label"},
    )
    hidden_entity_with_my_label = RegistryEntryWithDefaults(
        entity_id="light.hidden_with_my_label",
        unique_id="hidden_with_my_label",
        platform="test",
        labels={"my-label"},
        hidden_by=er.RegistryEntryHider.USER,
    )
    config_entity_with_my_label = RegistryEntryWithDefaults(
        entity_id="light.config_with_my_label",
        unique_id="config_with_my_label",
        platform="test",
        labels={"my-label"},
        entity_category=EntityCategory.CONFIG,
    )
    diag_entity_with_my_label = RegistryEntryWithDefaults(
        entity_id="light.diag_with_my_label",
        unique_id="diag_with_my_label",
        platform="test",
        labels={"my-label"},
        entity_category=EntityCategory.DIAGNOSTIC,
    )
    entity_with_label1_from_device = RegistryEntryWithDefaults(
        entity_id="light.with_label1_from_device",
        unique_id="with_label1_from_device",
        platform="test",
        device_id=device_has_label1.id,
    )
    entity_with_label1_from_device_and_different_area = RegistryEntryWithDefaults(
        entity_id="light.with_label1_from_device_diff_area",
        unique_id="with_label1_from_device_diff_area",
        platform="test",
        device_id=device_has_label1.id,
        area_id=area_without_labels.id,
    )
    entity_with_label1_and_label2_from_device = RegistryEntryWithDefaults(
        entity_id="light.with_label1_and_label2_from_device",
        unique_id="with_label1_and_label2_from_device",
        platform="test",
        labels={"label1"},
        device_id=device_has_label2.id,
    )
    entity_with_labels_from_device = RegistryEntryWithDefaults(
        entity_id="light.with_labels_from_device",
        unique_id="with_labels_from_device",
        platform="test",
        device_id=device_has_labels.id,
    )
    entity_with_no_labels = RegistryEntryWithDefaults(
        entity_id="light.no_labels",
        unique_id="no_labels",
        platform="test",
        device_id=device_no_labels.id,
    )

    mock_registry(
        hass,
        {
            config_entity_with_my_label.entity_id: config_entity_with_my_label,
            diag_entity_with_my_label.entity_id: diag_entity_with_my_label,
            entity_with_label1_and_label2_from_device.entity_id: entity_with_label1_and_label2_from_device,
            entity_with_label1_from_device.entity_id: entity_with_label1_from_device,
            entity_with_label1_from_device_and_different_area.entity_id: entity_with_label1_from_device_and_different_area,
            entity_with_labels_from_device.entity_id: entity_with_labels_from_device,
            entity_with_my_label.entity_id: entity_with_my_label,
            entity_with_no_labels.entity_id: entity_with_no_labels,
            hidden_entity_with_my_label.entity_id: hidden_entity_with_my_label,
        },
    )


async def test_call_from_config(hass: HomeAssistant) -> None:
    """Test the sync wrapper of service.async_call_from_config."""
    calls = async_mock_service(hass, "test_domain", "test_service")
    config = {
        "service_template": "{{ 'test_domain.test_service' }}",
        "entity_id": "hello.world",
        "data": {"hello": "goodbye"},
    }

    await hass.async_add_executor_job(service.call_from_config, hass, config)
    await hass.async_block_till_done()

    assert calls[0].data == {"hello": "goodbye", "entity_id": ["hello.world"]}


async def test_service_call(hass: HomeAssistant) -> None:
    """Test service call with templating."""
    calls = async_mock_service(hass, "test_domain", "test_service")
    config = {
        "action": "{{ 'test_domain.test_service' }}",
        "entity_id": "hello.world",
        "data": {
            "hello": "{{ 'goodbye' }}",
            "effect": {"value": "{{ 'complex' }}", "simple": "simple"},
        },
        "data_template": {"list": ["{{ 'list' }}", "2"]},
        "target": {
            "area_id": "test-area-id",
            "entity_id": "will.be_overridden",
            "floor_id": "test-floor-id",
        },
    }

    await service.async_call_from_config(hass, config)
    await hass.async_block_till_done()

    assert dict(calls[0].data) == {
        "hello": "goodbye",
        "effect": {
            "value": "complex",
            "simple": "simple",
        },
        "list": ["list", "2"],
        "entity_id": ["hello.world"],
        "area_id": ["test-area-id"],
        "floor_id": ["test-floor-id"],
    }

    # Templated strings in target fields
    config = {
        "action": "{{ 'test_domain.test_service' }}",
        "target": {
            "area_id": ["area-42", "{{ 'area-51' }}"],
            "device_id": ["abcdef", "{{ 'fedcba' }}"],
            "entity_id": ["light.static", "{{ 'light.dynamic' }}"],
            "floor_id": ["floor-first", "{{ 'floor-second' }}"],
        },
    }

    await service.async_call_from_config(hass, config)
    await hass.async_block_till_done()

    assert dict(calls[1].data) == {
        "area_id": ["area-42", "area-51"],
        "device_id": ["abcdef", "fedcba"],
        "entity_id": ["light.static", "light.dynamic"],
        "floor_id": ["floor-first", "floor-second"],
    }

    # Templated dict as target
    config = {
        "action": "{{ 'test_domain.test_service' }}",
        "target": "{{ var_target }}",
    }

    await service.async_call_from_config(
        hass,
        config,
        variables={
            "var_target": {
                "entity_id": "light.static",
                "area_id": ["area-42", "area-51"],
            },
        },
    )
    await hass.async_block_till_done()

    assert dict(calls[2].data) == {
        "area_id": ["area-42", "area-51"],
        "entity_id": ["light.static"],
    }

    # Templated lists as target fields
    config = {
        "action": "{{ 'test_domain.test_service' }}",
        "target": {
            "area_id": "{{ ['area-42', 'area-51'] }}",
            "device_id": "{{ ['abcdef', 'fedcba'] }}",
            "entity_id": "{{ ['light.static', 'light.dynamic'] }}",
            "floor_id": "{{ ['floor-first', 'floor-second'] }}",
        },
    }

    await service.async_call_from_config(hass, config)
    await hass.async_block_till_done()

    assert dict(calls[3].data) == {
        "area_id": ["area-42", "area-51"],
        "device_id": ["abcdef", "fedcba"],
        "entity_id": ["light.static", "light.dynamic"],
        "floor_id": ["floor-first", "floor-second"],
    }


async def test_service_template_service_call(hass: HomeAssistant) -> None:
    """Test legacy service_template call with templating."""
    calls = async_mock_service(hass, "test_domain", "test_service")
    config = {
        "service_template": "{{ 'test_domain.test_service' }}",
        "entity_id": "hello.world",
        "data": {"hello": "goodbye"},
    }

    await service.async_call_from_config(hass, config)
    await hass.async_block_till_done()

    assert calls[0].data == {"hello": "goodbye", "entity_id": ["hello.world"]}


async def test_passing_variables_to_templates(hass: HomeAssistant) -> None:
    """Test passing variables to templates."""
    calls = async_mock_service(hass, "test_domain", "test_service")
    config = {
        "service_template": "{{ var_service }}",
        "entity_id": "hello.world",
        "data_template": {"hello": "{{ var_data }}"},
    }

    await service.async_call_from_config(
        hass,
        config,
        variables={
            "var_service": "test_domain.test_service",
            "var_data": "goodbye",
        },
    )
    await hass.async_block_till_done()

    assert calls[0].data == {"hello": "goodbye", "entity_id": ["hello.world"]}


async def test_bad_template(hass: HomeAssistant) -> None:
    """Test passing bad template."""
    calls = async_mock_service(hass, "test_domain", "test_service")
    config = {
        "service_template": "{{ var_service }}",
        "entity_id": "hello.world",
        "data_template": {"hello": "{{ states + unknown_var }}"},
    }

    await service.async_call_from_config(
        hass,
        config,
        variables={
            "var_service": "test_domain.test_service",
            "var_data": "goodbye",
        },
    )
    await hass.async_block_till_done()

    assert len(calls) == 0


async def test_split_entity_string(hass: HomeAssistant) -> None:
    """Test splitting of entity string."""
    calls = async_mock_service(hass, "test_domain", "test_service")
    await service.async_call_from_config(
        hass,
        {
            "action": "test_domain.test_service",
            "entity_id": "hello.world, sensor.beer",
        },
    )
    await hass.async_block_till_done()
    assert calls[-1].data.get("entity_id") == ["hello.world", "sensor.beer"]


async def test_not_mutate_input(hass: HomeAssistant) -> None:
    """Test for immutable input."""
    async_mock_service(hass, "test_domain", "test_service")
    config = {
        "action": "test_domain.test_service",
        "entity_id": "hello.world, sensor.beer",
        "data": {"hello": 1},
        "data_template": {"nested": {"value": "{{ 1 + 1 }}"}},
    }
    orig = deepcopy(config)

    # Validate both the original and the copy
    config = cv.SERVICE_SCHEMA(config)
    orig = cv.SERVICE_SCHEMA(orig)

    await service.async_call_from_config(hass, config, validate_config=False)
    assert orig == config


@patch("homeassistant.helpers.service._LOGGER.error")
async def test_fail_silently_if_no_service(mock_log, hass: HomeAssistant) -> None:
    """Test failing if service is missing."""
    await service.async_call_from_config(hass, None)
    assert mock_log.call_count == 1

    await service.async_call_from_config(hass, {})
    assert mock_log.call_count == 2

    await service.async_call_from_config(hass, {"action": "invalid"})
    assert mock_log.call_count == 3


async def test_service_call_entry_id(
    hass: HomeAssistant, entity_registry: er.EntityRegistry
) -> None:
    """Test service call with entity specified by entity registry ID."""
    calls = async_mock_service(hass, "test_domain", "test_service")
    entry = entity_registry.async_get_or_create(
        "hello", "hue", "1234", suggested_object_id="world"
    )

    assert entry.entity_id == "hello.world"

    config = {
        "action": "test_domain.test_service",
        "target": {"entity_id": entry.id},
    }

    await service.async_call_from_config(hass, config)
    await hass.async_block_till_done()

    assert dict(calls[0].data) == {"entity_id": ["hello.world"]}


@pytest.mark.parametrize("target", ["all", "none"])
async def test_service_call_all_none(hass: HomeAssistant, target) -> None:
    """Test service call targeting all."""
    calls = async_mock_service(hass, "test_domain", "test_service")

    config = {
        "action": "test_domain.test_service",
        "target": {"entity_id": target},
    }

    await service.async_call_from_config(hass, config)
    await hass.async_block_till_done()

    assert dict(calls[0].data) == {"entity_id": target}


async def test_extract_entity_ids(hass: HomeAssistant) -> None:
    """Test extract_entity_ids method."""
    hass.states.async_set("light.Bowl", STATE_ON)
    hass.states.async_set("light.Ceiling", STATE_OFF)
    hass.states.async_set("light.Kitchen", STATE_OFF)

    assert await async_setup_component(hass, "group", {})
    await hass.async_block_till_done()
    await Group.async_create_group(
        hass,
        "test",
        created_by_service=False,
        entity_ids=["light.Ceiling", "light.Kitchen"],
        icon=None,
        mode=None,
        object_id=None,
        order=None,
    )

    call = ServiceCall(hass, "light", "turn_on", {ATTR_ENTITY_ID: "light.Bowl"})

    assert {"light.bowl"} == await service.async_extract_entity_ids(hass, call)

    call = ServiceCall(hass, "light", "turn_on", {ATTR_ENTITY_ID: "group.test"})

    assert {"light.ceiling", "light.kitchen"} == await service.async_extract_entity_ids(
        hass, call
    )

    assert {"group.test"} == await service.async_extract_entity_ids(
        hass, call, expand_group=False
    )

    assert (
        await service.async_extract_entity_ids(
            hass,
            ServiceCall(hass, "light", "turn_on", {ATTR_ENTITY_ID: ENTITY_MATCH_NONE}),
        )
        == set()
    )


async def test_extract_entity_ids_from_area(
    hass: HomeAssistant, floor_area_mock
) -> None:
    """Test extract_entity_ids method with areas."""
    call = ServiceCall(hass, "light", "turn_on", {"area_id": "own-area"})

    assert {
        "light.in_own_area",
    } == await service.async_extract_entity_ids(hass, call)

    call = ServiceCall(hass, "light", "turn_on", {"area_id": "test-area"})

    assert {
        "light.in_area",
        "light.assigned_to_area",
    } == await service.async_extract_entity_ids(hass, call)

    call = ServiceCall(
        hass, "light", "turn_on", {"area_id": ["test-area", "diff-area"]}
    )

    assert {
        "light.in_area",
        "light.diff_area",
        "light.assigned_to_area",
    } == await service.async_extract_entity_ids(hass, call)

    assert (
        await service.async_extract_entity_ids(
            hass, ServiceCall(hass, "light", "turn_on", {"area_id": ENTITY_MATCH_NONE})
        )
        == set()
    )


async def test_extract_entity_ids_from_devices(
    hass: HomeAssistant, floor_area_mock
) -> None:
    """Test extract_entity_ids method with devices."""
    assert await service.async_extract_entity_ids(
        hass, ServiceCall(hass, "light", "turn_on", {"device_id": "device-no-area-id"})
    ) == {
        "light.no_area",
    }

    assert await service.async_extract_entity_ids(
        hass, ServiceCall(hass, "light", "turn_on", {"device_id": "device-area-a-id"})
    ) == {
        "light.in_area_a",
        "light.in_area_b",
    }

    assert (
        await service.async_extract_entity_ids(
            hass,
            ServiceCall(hass, "light", "turn_on", {"device_id": "non-existing-id"}),
        )
        == set()
    )


@pytest.mark.usefixtures("floor_area_mock")
async def test_extract_entity_ids_from_floor(hass: HomeAssistant) -> None:
    """Test extract_entity_ids method with floors."""
    call = ServiceCall(hass, "light", "turn_on", {"floor_id": "test-floor"})

    assert {
        "light.in_area",
        "light.assigned_to_area",
    } == await service.async_extract_entity_ids(hass, call)

    call = ServiceCall(
        hass, "light", "turn_on", {"floor_id": ["test-floor", "floor-a"]}
    )

    assert {
        "light.in_area",
        "light.assigned_to_area",
        "light.in_area_a",
    } == await service.async_extract_entity_ids(hass, call)

    assert (
        await service.async_extract_entity_ids(
            hass, ServiceCall(hass, "light", "turn_on", {"floor_id": ENTITY_MATCH_NONE})
        )
        == set()
    )


@pytest.mark.usefixtures("label_mock")
async def test_extract_entity_ids_from_labels(hass: HomeAssistant) -> None:
    """Test extract_entity_ids method with labels."""
    call = ServiceCall(hass, "light", "turn_on", {"label_id": "my-label"})

    assert {
        "light.with_my_label",
        "light.config_with_my_label",
        "light.diag_with_my_label",
    } == await service.async_extract_entity_ids(hass, call)

    call = ServiceCall(hass, "light", "turn_on", {"label_id": "label1"})

    assert {
        "light.with_label1_from_device",
        "light.with_label1_from_device_diff_area",
        "light.with_labels_from_device",
        "light.with_label1_and_label2_from_device",
    } == await service.async_extract_entity_ids(hass, call)

    call = ServiceCall(hass, "light", "turn_on", {"label_id": ["label2"]})

    assert {
        "light.with_labels_from_device",
        "light.with_label1_and_label2_from_device",
    } == await service.async_extract_entity_ids(hass, call)

    call = ServiceCall(hass, "light", "turn_on", {"label_id": ["label_area"]})

    assert {
        "light.with_labels_from_device",
    } == await service.async_extract_entity_ids(hass, call)

    assert (
        await service.async_extract_entity_ids(
            hass, ServiceCall(hass, "light", "turn_on", {"label_id": ENTITY_MATCH_NONE})
        )
        == set()
    )


async def test_async_get_all_descriptions(hass: HomeAssistant) -> None:
    """Test async_get_all_descriptions."""
    group_config = {DOMAIN_GROUP: {}}
    assert await async_setup_component(hass, DOMAIN_GROUP, group_config)
    assert await async_setup_component(hass, DOMAIN_SYSTEM_HEALTH, {})

    with patch(
        "homeassistant.helpers.service._load_services_files",
        side_effect=service._load_services_files,
    ) as proxy_load_services_files:
        descriptions = await service.async_get_all_descriptions(hass)

    # Test we only load services.yaml for integrations with services.yaml
    # And system_health has no services
    assert proxy_load_services_files.mock_calls[0][1][0] == unordered(
        [
            await async_get_integration(hass, DOMAIN_GROUP),
        ]
    )

    assert len(descriptions) == 1
    assert DOMAIN_GROUP in descriptions
    assert "description" not in descriptions[DOMAIN_GROUP]["reload"]
    assert "fields" in descriptions[DOMAIN_GROUP]["reload"]

    # Does not have services
    assert DOMAIN_SYSTEM_HEALTH not in descriptions

    logger_config = {DOMAIN_LOGGER: {}}

    # Test legacy service with translations in services.yaml
    def _load_services_file(integration: Integration) -> JSON_TYPE:
        return {
            "set_default_level": {
                "description": "Translated description",
                "fields": {
                    "level": {
                        "description": "Field description",
                        "example": "Field example",
                        "name": "Field name",
                        "selector": {
                            "select": {
                                "options": [
                                    "debug",
                                    "info",
                                    "warning",
                                    "error",
                                    "fatal",
                                    "critical",
                                ],
                                "translation_key": "level",
                            }
                        },
                    }
                },
                "name": "Translated name",
            },
            "set_level": None,
        }

    with patch(
        "homeassistant.helpers.service._load_services_file",
        side_effect=_load_services_file,
    ):
        await async_setup_component(hass, DOMAIN_LOGGER, logger_config)
        descriptions = await service.async_get_all_descriptions(hass)

    assert len(descriptions) == 2
    assert DOMAIN_LOGGER in descriptions
    assert descriptions[DOMAIN_LOGGER]["set_default_level"]["name"] == "Translated name"
    assert (
        descriptions[DOMAIN_LOGGER]["set_default_level"]["description"]
        == "Translated description"
    )
    assert (
        descriptions[DOMAIN_LOGGER]["set_default_level"]["fields"]["level"]["name"]
        == "Field name"
    )
    assert (
        descriptions[DOMAIN_LOGGER]["set_default_level"]["fields"]["level"][
            "description"
        ]
        == "Field description"
    )
    assert (
        descriptions[DOMAIN_LOGGER]["set_default_level"]["fields"]["level"]["example"]
        == "Field example"
    )

    hass.services.async_register(DOMAIN_LOGGER, "new_service", lambda x: None, None)
    service.async_set_service_schema(
        hass, DOMAIN_LOGGER, "new_service", {"description": "new service"}
    )
    descriptions = await service.async_get_all_descriptions(hass)
    assert "description" in descriptions[DOMAIN_LOGGER]["new_service"]
    assert descriptions[DOMAIN_LOGGER]["new_service"]["description"] == "new service"

    hass.services.async_register(
        DOMAIN_LOGGER, "another_new_service", lambda x: None, None
    )
    hass.services.async_register(
        DOMAIN_LOGGER,
        "service_with_optional_response",
        lambda x: None,
        None,
        SupportsResponse.OPTIONAL,
    )
    hass.services.async_register(
        DOMAIN_LOGGER,
        "service_with_only_response",
        lambda x: None,
        None,
        SupportsResponse.ONLY,
    )
    hass.services.async_register(
        DOMAIN_LOGGER,
        "another_service_with_response",
        lambda x: None,
        None,
        SupportsResponse.OPTIONAL,
    )
    service.async_set_service_schema(
        hass,
        DOMAIN_LOGGER,
        "another_service_with_response",
        {"description": "response service"},
    )
    descriptions = await service.async_get_all_descriptions(hass)
    assert "another_new_service" in descriptions[DOMAIN_LOGGER]
    assert "service_with_optional_response" in descriptions[DOMAIN_LOGGER]
    assert descriptions[DOMAIN_LOGGER]["service_with_optional_response"][
        "response"
    ] == {"optional": True}
    assert "service_with_only_response" in descriptions[DOMAIN_LOGGER]
    assert descriptions[DOMAIN_LOGGER]["service_with_only_response"]["response"] == {
        "optional": False
    }
    assert "another_service_with_response" in descriptions[DOMAIN_LOGGER]
    assert descriptions[DOMAIN_LOGGER]["another_service_with_response"]["response"] == {
        "optional": True
    }

    # Verify the cache returns the same object
    assert await service.async_get_all_descriptions(hass) is descriptions


async def test_async_get_all_descriptions_dot_keys(hass: HomeAssistant) -> None:
    """Test async_get_all_descriptions with keys starting with a period."""
    service_descriptions = """
        .anchor: &anchor
          selector:
            text:
        test_service:
          fields:
            test: *anchor
    """

    domain = "test_domain"

    hass.services.async_register(domain, "test_service", lambda call: None)
    mock_integration(hass, MockModule(domain), top_level_files={"services.yaml"})
    assert await async_setup_component(hass, domain, {})

    def load_yaml(fname, secrets=None):
        with io.StringIO(service_descriptions) as file:
            return parse_yaml(file)

    with (
        patch(
            "homeassistant.helpers.service._load_services_files",
            side_effect=service._load_services_files,
        ) as proxy_load_services_files,
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=load_yaml,
        ) as mock_load_yaml,
    ):
        descriptions = await service.async_get_all_descriptions(hass)

    mock_load_yaml.assert_called_once_with(
        "homeassistant/components/test_domain/services.yaml", None
    )
    assert proxy_load_services_files.mock_calls[0][1][0] == unordered(
        [
            await async_get_integration(hass, domain),
        ]
    )

    assert descriptions == {
        "test_domain": {
            "test_service": {
                "fields": {
                    "test": {
                        "selector": {"text": {"multiline": False, "multiple": False}}
                    }
                },
            }
        }
    }


async def test_async_get_all_descriptions_filter(hass: HomeAssistant) -> None:
    """Test async_get_all_descriptions with filters."""
    service_descriptions = """
        test_service:
          target:
            entity:
              domain: alarm_control_panel
              supported_features:
                - alarm_control_panel.AlarmControlPanelEntityFeature.ARM_HOME
          fields:
            temperature:
              filter:
                supported_features:
                  - alarm_control_panel.AlarmControlPanelEntityFeature.ARM_HOME
                attribute:
                  supported_color_modes:
                    - light.ColorMode.COLOR_TEMP
              selector:
                number:
            entity:
              selector:
                entity:
                  filter:
                    domain: alarm_control_panel
                    supported_features:
                      - alarm_control_panel.AlarmControlPanelEntityFeature.ARM_HOME
            advanced_stuff:
              fields:
                temperature:
                  filter:
                    supported_features:
                      - alarm_control_panel.AlarmControlPanelEntityFeature.ARM_HOME
                    attribute:
                      supported_color_modes:
                        - light.ColorMode.COLOR_TEMP
                  selector:
                    number:
                entity:
                  selector:
                    entity:
                      filter:
                        domain: alarm_control_panel
                        supported_features:
                          - alarm_control_panel.AlarmControlPanelEntityFeature.ARM_HOME
    """

    domain = "test_domain"

    hass.services.async_register(domain, "test_service", lambda call: None)
    mock_integration(hass, MockModule(domain), top_level_files={"services.yaml"})
    assert await async_setup_component(hass, domain, {})

    def load_yaml(fname, secrets=None):
        with io.StringIO(service_descriptions) as file:
            return parse_yaml(file)

    with (
        patch(
            "homeassistant.helpers.service._load_services_files",
            side_effect=service._load_services_files,
        ) as proxy_load_services_files,
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=load_yaml,
        ) as mock_load_yaml,
    ):
        descriptions = await service.async_get_all_descriptions(hass)

    mock_load_yaml.assert_called_once_with(
        "homeassistant/components/test_domain/services.yaml", None
    )
    assert proxy_load_services_files.mock_calls[0][1][0] == unordered(
        [
            await async_get_integration(hass, domain),
        ]
    )

    test_service_schema = {
        "fields": {
            "advanced_stuff": {
                "fields": {
                    "temperature": {
                        "filter": {
                            "attribute": {"supported_color_modes": ["color_temp"]},
                            "supported_features": [1],
                        },
                        "selector": {
                            "number": {
                                "mode": "box",
                                "step": 1.0,
                            }
                        },
                    },
                    "entity": {
                        "selector": {
                            "entity": {
                                "filter": [
                                    {
                                        "domain": ["alarm_control_panel"],
                                        "supported_features": [1],
                                    }
                                ],
                                "multiple": False,
                                "reorder": False,
                            },
                        },
                    },
                },
            },
            "temperature": {
                "filter": {
                    "attribute": {"supported_color_modes": ["color_temp"]},
                    "supported_features": [1],
                },
                "selector": {
                    "number": {
                        "mode": "box",
                        "step": 1.0,
                    }
                },
            },
            "entity": {
                "selector": {
                    "entity": {
                        "filter": [
                            {
                                "domain": ["alarm_control_panel"],
                                "supported_features": [1],
                            }
                        ],
                        "multiple": False,
                        "reorder": False,
                    },
                },
            },
        },
        "target": {
            "entity": [
                {
                    "domain": ["alarm_control_panel"],
                    "supported_features": [1],
                },
            ],
        },
    }

    assert descriptions == {
        "test_domain": {"test_service": test_service_schema},
    }


async def test_async_get_all_descriptions_failing_integration(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test async_get_all_descriptions when async_get_integrations returns an exception."""
    group_config = {DOMAIN_GROUP: {}}
    await async_setup_component(hass, DOMAIN_GROUP, group_config)

    logger_config = {DOMAIN_LOGGER: {}}
    await async_setup_component(hass, DOMAIN_LOGGER, logger_config)

    input_button_config = {DOMAIN_INPUT_BUTTON: {}}
    await async_setup_component(hass, DOMAIN_INPUT_BUTTON, input_button_config)

    async def wrap_get_integrations(
        hass: HomeAssistant, domains: Iterable[str]
    ) -> dict[str, Integration | Exception]:
        integrations = await async_get_integrations(hass, domains)
        integrations[DOMAIN_LOGGER] = ImportError("Failed to load services.yaml")
        return integrations

    with (
        patch(
            "homeassistant.helpers.service.async_get_integrations",
            wraps=wrap_get_integrations,
        ),
    ):
        descriptions = await service.async_get_all_descriptions(hass)

    assert len(descriptions) == 3
    assert "Failed to load services.yaml for integration: logger" in caplog.text

    # Services are empty defaults if the load fails but should
    # not raise
    assert "description" not in descriptions[DOMAIN_GROUP]["remove"]
    assert descriptions[DOMAIN_GROUP]["remove"]["fields"]

    assert descriptions[DOMAIN_LOGGER]["set_level"] == {"fields": {}}

    assert "description" not in descriptions[DOMAIN_INPUT_BUTTON]["press"]
    assert descriptions[DOMAIN_INPUT_BUTTON]["press"]["fields"] == {}
    assert "target" in descriptions[DOMAIN_INPUT_BUTTON]["press"]

    hass.services.async_register(DOMAIN_LOGGER, "new_service", lambda x: None, None)
    service.async_set_service_schema(
        hass, DOMAIN_LOGGER, "new_service", {"description": "new service"}
    )
    descriptions = await service.async_get_all_descriptions(hass)
    assert "description" in descriptions[DOMAIN_LOGGER]["new_service"]
    assert descriptions[DOMAIN_LOGGER]["new_service"]["description"] == "new service"

    hass.services.async_register(
        DOMAIN_LOGGER, "another_new_service", lambda x: None, None
    )
    hass.services.async_register(
        DOMAIN_LOGGER,
        "service_with_optional_response",
        lambda x: None,
        None,
        SupportsResponse.OPTIONAL,
    )
    hass.services.async_register(
        DOMAIN_LOGGER,
        "service_with_only_response",
        lambda x: None,
        None,
        SupportsResponse.ONLY,
    )

    descriptions = await service.async_get_all_descriptions(hass)
    assert "another_new_service" in descriptions[DOMAIN_LOGGER]
    assert "service_with_optional_response" in descriptions[DOMAIN_LOGGER]
    assert descriptions[DOMAIN_LOGGER]["service_with_optional_response"][
        "response"
    ] == {"optional": True}
    assert "service_with_only_response" in descriptions[DOMAIN_LOGGER]
    assert descriptions[DOMAIN_LOGGER]["service_with_only_response"]["response"] == {
        "optional": False
    }

    # Verify the cache returns the same object
    assert await service.async_get_all_descriptions(hass) is descriptions


async def test_async_get_all_descriptions_dynamically_created_services(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test async_get_all_descriptions when async_get_integrations when services are dynamic."""
    group_config = {DOMAIN_GROUP: {}}
    await async_setup_component(hass, DOMAIN_GROUP, group_config)
    descriptions = await service.async_get_all_descriptions(hass)

    assert len(descriptions) == 1

    assert "description" not in descriptions["group"]["reload"]
    assert "fields" in descriptions["group"]["reload"]

    shell_command_config = {DOMAIN_SHELL_COMMAND: {"test_service": "ls /bin"}}
    await async_setup_component(hass, DOMAIN_SHELL_COMMAND, shell_command_config)
    descriptions = await service.async_get_all_descriptions(hass)

    assert len(descriptions) == 2
    assert descriptions[DOMAIN_SHELL_COMMAND]["test_service"] == {
        "fields": {},
        "response": {"optional": True},
    }


async def test_async_get_all_descriptions_new_service_added_while_loading(
    hass: HomeAssistant,
) -> None:
    """Test async_get_all_descriptions when a new service is added while loading translations."""
    group_config = {DOMAIN_GROUP: {}}
    await async_setup_component(hass, DOMAIN_GROUP, group_config)
    descriptions = await service.async_get_all_descriptions(hass)

    assert len(descriptions) == 1

    assert "description" not in descriptions["group"]["reload"]
    assert "fields" in descriptions["group"]["reload"]

    logger_domain = DOMAIN_LOGGER
    logger_config = {logger_domain: {}}

    translations_called = threading.Event()
    translations_wait = threading.Event()

    def _load_services_file(integration: Integration) -> JSON_TYPE:
        translations_called.set()
        translations_wait.wait()
        return {
            "set_default_level": {
                "description": "Translated description",
                "fields": {
                    "level": {
                        "description": "Field description",
                        "example": "Field example",
                        "name": "Field name",
                        "selector": {
                            "select": {
                                "options": [
                                    "debug",
                                    "info",
                                    "warning",
                                    "error",
                                    "fatal",
                                    "critical",
                                ],
                                "translation_key": "level",
                            }
                        },
                    }
                },
                "name": "Translated name",
            },
            "set_level": None,
        }

    with patch(
        "homeassistant.helpers.service._load_services_file",
        side_effect=_load_services_file,
    ):
        await async_setup_component(hass, logger_domain, logger_config)
        task = asyncio.create_task(service.async_get_all_descriptions(hass))
        await hass.async_add_executor_job(translations_called.wait)
        # Now register a new service while translations are being loaded
        hass.services.async_register(logger_domain, "new_service", lambda x: None, None)
        service.async_set_service_schema(
            hass, logger_domain, "new_service", {"description": "new service"}
        )
        translations_wait.set()
        descriptions = await task

    # Two domains should be present
    assert len(descriptions) == 2

    logger_descriptions = descriptions[logger_domain]

    # The new service was loaded after the translations were loaded
    # so it should not appear until the next time we fetch
    assert "new_service" not in logger_descriptions

    set_default_level = logger_descriptions["set_default_level"]

    assert set_default_level["name"] == "Translated name"
    assert set_default_level["description"] == "Translated description"
    set_default_level_fields = set_default_level["fields"]
    assert set_default_level_fields["level"]["name"] == "Field name"
    assert set_default_level_fields["level"]["description"] == "Field description"
    assert set_default_level_fields["level"]["example"] == "Field example"

    descriptions = await service.async_get_all_descriptions(hass)
    assert "description" in descriptions[logger_domain]["new_service"]
    assert descriptions[logger_domain]["new_service"]["description"] == "new service"


async def test_async_get_descriptions_with_placeholders(hass: HomeAssistant) -> None:
    """Test descriptions async_get_all_descriptions with placeholders.

    Placeholders supplied with a service registration should be included.
    """
    service_descriptions = """
    happy_time:
      fields:
        topic:
          selector:
            text:
        duration:
          default: 5
          selector:
            number:
              min: 1
              max: 300
              unit_of_measurement: "seconds"
    """

    service_schema = vol.Schema(
        {
            "topic": cv.string,
            "duration": cv.positive_int,
        }
    )

    domain = "test_domain"

    hass.services.async_register(
        domain,
        "happy_time",
        lambda call: None,
        schema=service_schema,
        description_placeholders={"placeholder": "beer"},
    )
    mock_integration(hass, MockModule(domain), top_level_files={"services.yaml"})
    assert await async_setup_component(hass, domain, {})

    def load_yaml(fname, secrets=None):
        with io.StringIO(service_descriptions) as file:
            return parse_yaml(file)

    with (
        patch(
            "homeassistant.helpers.service._load_services_files",
            side_effect=service._load_services_files,
        ) as proxy_load_services_files,
        patch(
            "annotatedyaml.loader.load_yaml",
            side_effect=load_yaml,
        ) as mock_load_yaml,
    ):
        descriptions = await service.async_get_all_descriptions(hass)

    mock_load_yaml.assert_called_once_with(
        "homeassistant/components/test_domain/services.yaml", None
    )
    assert proxy_load_services_files.mock_calls[0][1][0] == unordered(
        [
            await async_get_integration(hass, domain),
        ]
    )

    assert descriptions == {
        "test_domain": {
            "happy_time": {
                "fields": {
                    "topic": {
                        "selector": {"text": {"multiple": False, "multiline": False}}
                    },
                    "duration": {
                        "default": 5,
                        "selector": {
                            "number": {
                                "min": 1.0,
                                "max": 300.0,
                                "unit_of_measurement": "seconds",
                                "step": 1.0,
                                "mode": "slider",
                            }
                        },
                    },
                },
                "description_placeholders": {"placeholder": "beer"},
            }
        }
    }


async def test_register_with_mixed_case(hass: HomeAssistant) -> None:
    """Test registering a service with mixed case.

    For backwards compatibility, we have historically allowed mixed case,
    and automatically converted it to lowercase.
    """
    logger_config = {DOMAIN_LOGGER: {}}
    await async_setup_component(hass, DOMAIN_LOGGER, logger_config)
    logger_domain_mixed = "LoGgEr"
    hass.services.async_register(
        logger_domain_mixed, "NeW_SeRVICE", lambda x: None, None
    )
    service.async_set_service_schema(
        hass, logger_domain_mixed, "NeW_SeRVICE", {"description": "new service"}
    )
    descriptions = await service.async_get_all_descriptions(hass)
    assert "description" in descriptions[DOMAIN_LOGGER]["new_service"]
    assert descriptions[DOMAIN_LOGGER]["new_service"]["description"] == "new service"


async def test_call_with_required_features(hass: HomeAssistant, mock_entities) -> None:
    """Test service calls invoked only if entity has required features."""
    # Set up homeassistant component to fetch the translations
    await async_setup_component(hass, "homeassistant", {})
    test_service_mock = AsyncMock(return_value=None)
    await service.entity_service_call(
        hass,
        mock_entities,
        HassJob(test_service_mock),
        ServiceCall(hass, "test_domain", "test_service", {"entity_id": "all"}),
        required_features=[SUPPORT_A],
    )

    assert test_service_mock.call_count == 2
    expected = [
        mock_entities["light.kitchen"],
        mock_entities["light.bedroom"],
    ]
    actual = [call[0][0] for call in test_service_mock.call_args_list]
    assert actual == unordered(expected)

    # Test we raise if we target entity ID that does not support the service
    test_service_mock.reset_mock()
    with pytest.raises(
        exceptions.ServiceNotSupported,
        match="Entity light.living_room does not "
        "support action test_domain.test_service",
    ):
        await service.entity_service_call(
            hass,
            mock_entities,
            HassJob(test_service_mock),
            ServiceCall(
                hass, "test_domain", "test_service", {"entity_id": "light.living_room"}
            ),
            required_features=[SUPPORT_A],
        )
    assert test_service_mock.call_count == 0


async def test_call_with_both_required_features(
    hass: HomeAssistant, mock_entities
) -> None:
    """Test service calls invoked only if entity has both features."""
    test_service_mock = AsyncMock(return_value=None)
    await service.entity_service_call(
        hass,
        mock_entities,
        HassJob(test_service_mock),
        ServiceCall(hass, "test_domain", "test_service", {"entity_id": "all"}),
        required_features=[SUPPORT_A | SUPPORT_B],
    )

    assert test_service_mock.call_count == 1
    assert [call[0][0] for call in test_service_mock.call_args_list] == [
        mock_entities["light.bedroom"]
    ]


async def test_call_with_one_of_required_features(
    hass: HomeAssistant, mock_entities
) -> None:
    """Test service calls invoked with one entity having the required features."""
    test_service_mock = AsyncMock(return_value=None)
    await service.entity_service_call(
        hass,
        mock_entities,
        HassJob(test_service_mock),
        ServiceCall(hass, "test_domain", "test_service", {"entity_id": "all"}),
        required_features=[SUPPORT_A, SUPPORT_C],
    )

    assert test_service_mock.call_count == 3
    expected = [
        mock_entities["light.kitchen"],
        mock_entities["light.bedroom"],
        mock_entities["light.bathroom"],
    ]
    actual = [call[0][0] for call in test_service_mock.call_args_list]
    assert actual == unordered(expected)


@pytest.mark.parametrize(
    ("entity_device_classes", "expected_entities", "unsupported_entity"),
    [
        (
            [None],
            ["light.kitchen"],
            "light.living_room",
        ),
        (
            ["class_a"],
            ["light.living_room"],
            "light.kitchen",
        ),
        (
            [None, "class_a"],
            ["light.kitchen", "light.living_room"],
            "light.bedroom",
        ),
    ],
)
async def test_call_with_device_class(
    hass: HomeAssistant,
    mock_entities,
    entity_device_classes: list[str | None],
    expected_entities: list[str],
    unsupported_entity: str,
) -> None:
    """Test service calls invoked only if entity has required features."""
    # Set up homeassistant component to fetch the translations
    await async_setup_component(hass, "homeassistant", {})
    test_service_mock = AsyncMock(return_value=None)
    await service.entity_service_call(
        hass,
        mock_entities,
        HassJob(test_service_mock),
        ServiceCall(hass, "test_domain", "test_service", {"entity_id": "all"}),
        entity_device_classes=entity_device_classes,
    )

    assert test_service_mock.call_count == len(expected_entities)
    expected = [mock_entities[expected_entity] for expected_entity in expected_entities]
    actual = [call[0][0] for call in test_service_mock.call_args_list]
    assert actual == unordered(expected)

    # Test we raise if we target entity ID that does not support the service
    test_service_mock.reset_mock()
    with pytest.raises(
        exceptions.ServiceNotSupported,
        match=f"Entity {unsupported_entity} does not "
        "support action test_domain.test_service",
    ):
        await service.entity_service_call(
            hass,
            mock_entities,
            HassJob(test_service_mock),
            ServiceCall(
                hass, "test_domain", "test_service", {"entity_id": unsupported_entity}
            ),
            entity_device_classes=entity_device_classes,
        )
    assert test_service_mock.call_count == 0


async def test_call_with_sync_func(hass: HomeAssistant, mock_entities) -> None:
    """Test invoking sync service calls."""
    test_service_mock = Mock(return_value=None)
    await service.entity_service_call(
        hass,
        mock_entities,
        HassJob(test_service_mock),
        ServiceCall(
            hass, "test_domain", "test_service", {"entity_id": "light.kitchen"}
        ),
    )
    assert test_service_mock.call_count == 1


async def test_call_with_sync_attr(hass: HomeAssistant, mock_entities) -> None:
    """Test invoking sync service calls."""
    mock_method = mock_entities["light.kitchen"].sync_method = Mock(return_value=None)
    await service.entity_service_call(
        hass,
        mock_entities,
        "sync_method",
        ServiceCall(
            hass,
            "test_domain",
            "test_service",
            {"entity_id": "light.kitchen", "area_id": "abcd"},
        ),
    )
    assert mock_method.call_count == 1
    # We pass empty kwargs because both entity_id and area_id are filtered out
    assert mock_method.mock_calls[0][2] == {}


async def test_call_context_user_not_exist(hass: HomeAssistant) -> None:
    """Check we don't allow deleted users to do things."""
    with pytest.raises(exceptions.UnknownUser) as err:
        await service.entity_service_call(
            hass,
            {},
            Mock(),
            ServiceCall(
                hass,
                "test_domain",
                "test_service",
                context=Context(user_id="non-existing"),
            ),
        )

    assert err.value.context.user_id == "non-existing"


async def test_call_context_target_all(
    hass: HomeAssistant, mock_handle_entity_call, mock_entities
) -> None:
    """Check we only target allowed entities if targeting all."""
    with patch(
        "homeassistant.auth.AuthManager.async_get_user",
        return_value=Mock(
            permissions=PolicyPermissions(
                {"entities": {"entity_ids": {"light.kitchen": True}}}, None
            ),
            is_admin=False,
        ),
    ):
        await service.entity_service_call(
            hass,
            mock_entities,
            Mock(),
            ServiceCall(
                hass,
                "test_domain",
                "test_service",
                data={"entity_id": ENTITY_MATCH_ALL},
                context=Context(user_id="mock-id"),
            ),
        )

    assert len(mock_handle_entity_call.mock_calls) == 1
    assert mock_handle_entity_call.mock_calls[0][1][1].entity_id == "light.kitchen"


async def test_call_context_target_specific(
    hass: HomeAssistant, mock_handle_entity_call, mock_entities
) -> None:
    """Check targeting specific entities."""
    with patch(
        "homeassistant.auth.AuthManager.async_get_user",
        return_value=Mock(
            permissions=PolicyPermissions(
                {"entities": {"entity_ids": {"light.kitchen": True}}}, None
            )
        ),
    ):
        await service.entity_service_call(
            hass,
            mock_entities,
            Mock(),
            ServiceCall(
                hass,
                "test_domain",
                "test_service",
                {"entity_id": "light.kitchen"},
                context=Context(user_id="mock-id"),
            ),
        )

    assert len(mock_handle_entity_call.mock_calls) == 1
    assert mock_handle_entity_call.mock_calls[0][1][1].entity_id == "light.kitchen"


async def test_call_context_target_specific_no_auth(
    hass: HomeAssistant, mock_handle_entity_call, mock_entities
) -> None:
    """Check targeting specific entities without auth."""
    with (
        pytest.raises(exceptions.Unauthorized) as err,
        patch(
            "homeassistant.auth.AuthManager.async_get_user",
            return_value=Mock(permissions=PolicyPermissions({}, None), is_admin=False),
        ),
    ):
        await service.entity_service_call(
            hass,
            mock_entities,
            Mock(),
            ServiceCall(
                hass,
                "test_domain",
                "test_service",
                {"entity_id": "light.kitchen"},
                context=Context(user_id="mock-id"),
            ),
        )

    assert err.value.context.user_id == "mock-id"
    assert err.value.entity_id == "light.kitchen"


async def test_call_no_context_target_all(
    hass: HomeAssistant, mock_handle_entity_call, mock_entities
) -> None:
    """Check we target all if no user context given."""
    await service.entity_service_call(
        hass,
        mock_entities,
        Mock(),
        ServiceCall(
            hass, "test_domain", "test_service", data={"entity_id": ENTITY_MATCH_ALL}
        ),
    )

    assert len(mock_handle_entity_call.mock_calls) == 4
    assert [call[1][1] for call in mock_handle_entity_call.mock_calls] == list(
        mock_entities.values()
    )


async def test_call_no_context_target_specific(
    hass: HomeAssistant, mock_handle_entity_call, mock_entities
) -> None:
    """Check we can target specified entities."""
    await service.entity_service_call(
        hass,
        mock_entities,
        Mock(),
        ServiceCall(
            hass,
            "test_domain",
            "test_service",
            {"entity_id": ["light.kitchen", "light.non-existing"]},
        ),
    )

    assert len(mock_handle_entity_call.mock_calls) == 1
    assert mock_handle_entity_call.mock_calls[0][1][1].entity_id == "light.kitchen"


async def test_call_with_match_all(
    hass: HomeAssistant,
    mock_handle_entity_call,
    mock_entities,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Check we only target allowed entities if targeting all."""
    await service.entity_service_call(
        hass,
        mock_entities,
        Mock(),
        ServiceCall(hass, "test_domain", "test_service", {"entity_id": "all"}),
    )

    assert len(mock_handle_entity_call.mock_calls) == 4
    assert [call[1][1] for call in mock_handle_entity_call.mock_calls] == list(
        mock_entities.values()
    )


async def test_call_with_omit_entity_id(
    hass: HomeAssistant, mock_handle_entity_call, mock_entities
) -> None:
    """Check service call if we do not pass an entity ID."""
    await service.entity_service_call(
        hass,
        mock_entities,
        Mock(),
        ServiceCall(hass, "test_domain", "test_service"),
    )

    assert len(mock_handle_entity_call.mock_calls) == 0


async def test_register_admin_service(
    hass: HomeAssistant, hass_read_only_user: MockUser, hass_admin_user: MockUser
) -> None:
    """Test the register admin service."""
    calls = []

    async def mock_service(call):
        calls.append(call)

    service.async_register_admin_service(hass, "test", "test", mock_service)
    service.async_register_admin_service(
        hass,
        "test",
        "test2",
        mock_service,
        vol.Schema({vol.Required("required"): cv.boolean}),
    )

    with pytest.raises(exceptions.UnknownUser):
        await hass.services.async_call(
            "test",
            "test",
            {},
            blocking=True,
            context=Context(user_id="non-existing"),
        )
    assert len(calls) == 0

    with pytest.raises(exceptions.Unauthorized):
        await hass.services.async_call(
            "test",
            "test",
            {},
            blocking=True,
            context=Context(user_id=hass_read_only_user.id),
        )
    assert len(calls) == 0

    with pytest.raises(vol.Invalid):
        await hass.services.async_call(
            "test",
            "test",
            {"invalid": True},
            blocking=True,
            context=Context(user_id=hass_admin_user.id),
        )
    assert len(calls) == 0

    with pytest.raises(vol.Invalid):
        await hass.services.async_call(
            "test",
            "test2",
            {},
            blocking=True,
            context=Context(user_id=hass_admin_user.id),
        )
    assert len(calls) == 0

    await hass.services.async_call(
        "test",
        "test2",
        {"required": True},
        blocking=True,
        context=Context(user_id=hass_admin_user.id),
    )
    assert len(calls) == 1
    assert calls[0].context.user_id == hass_admin_user.id


async def test_register_admin_service_with_placeholders(
    hass: HomeAssistant, hass_admin_user: MockUser
) -> None:
    """Test the register admin service with description placeholders."""
    calls = []

    async def mock_service(call):
        calls.append(call)

    service.async_register_admin_service(
        hass,
        "test",
        "test",
        mock_service,
        description_placeholders={"test_placeholder": "beer"},
    )
    await hass.services.async_call(
        "test",
        "test",
        {},
        blocking=True,
        context=Context(user_id=hass_admin_user.id),
    )
    assert len(calls) == 1

    descriptions = await service.async_get_all_descriptions(hass)
    assert descriptions["test"]["test"]["description_placeholders"] == {
        "test_placeholder": "beer"
    }


@pytest.mark.parametrize(
    "supports_response",
    [SupportsResponse.ONLY, SupportsResponse.OPTIONAL],
)
async def test_register_admin_service_return_response(
    hass: HomeAssistant, supports_response: SupportsResponse
) -> None:
    """Test the register admin service for a service that returns response data."""

    async def mock_service(call: ServiceCall) -> ServiceResponse:
        """Service handler coroutine."""
        assert call.return_response
        return {"test-reply": "test-value1"}

    service.async_register_admin_service(
        hass, "test", "test", mock_service, supports_response=supports_response
    )
    result = await hass.services.async_call(
        "test",
        "test",
        service_data={},
        blocking=True,
        return_response=True,
    )
    assert result == {"test-reply": "test-value1"}


_DEPRECATED_VERIFY_DOMAIN_CONTROL_MESSAGE = (
    "The deprecated argument hass was passed to verify_domain_control. It will be"
    " removed in HA Core 2026.10. Use verify_domain_control without hass argument"
    " instead"
)


@pytest.mark.parametrize(
    # Check that with or without hass behaves the same
    ("decorator", "in_caplog"),
    [
        (service.verify_domain_control, True),  # old pass-through
        (lambda _, domain: service.verify_domain_control(domain), False),  # new
    ],
)
async def test_domain_control_not_async(
    hass: HomeAssistant,
    mock_entities,
    decorator: Callable[[HomeAssistant, str], Any],
    in_caplog: bool,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test domain verification in a service call with an unknown user."""
    calls = []

    def mock_service_log(call):
        """Define a protected service."""
        calls.append(call)

    with pytest.raises(exceptions.HomeAssistantError):
        decorator(hass, "test_domain")(mock_service_log)

    assert (_DEPRECATED_VERIFY_DOMAIN_CONTROL_MESSAGE in caplog.text) == in_caplog


@pytest.mark.parametrize(
    # Check that with or without hass behaves the same
    ("decorator", "in_caplog"),
    [
        (service.verify_domain_control, True),  # old pass-through
        (lambda _, domain: service.verify_domain_control(domain), False),  # new
    ],
)
async def test_domain_control_unknown(
    hass: HomeAssistant,
    mock_entities,
    decorator: Callable[[HomeAssistant, str], Any],
    in_caplog: bool,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test domain verification in a service call with an unknown user."""
    calls = []

    async def mock_service_log(call):
        """Define a protected service."""
        calls.append(call)

    with patch(
        "homeassistant.helpers.entity_registry.async_get",
        return_value=Mock(entities=mock_entities),
    ):
        protected_mock_service = decorator(hass, "test_domain")(mock_service_log)

        hass.services.async_register(
            "test_domain", "test_service", protected_mock_service, schema=None
        )

        with pytest.raises(exceptions.UnknownUser):
            await hass.services.async_call(
                "test_domain",
                "test_service",
                {},
                blocking=True,
                context=Context(user_id="fake_user_id"),
            )
        assert len(calls) == 0

    assert (_DEPRECATED_VERIFY_DOMAIN_CONTROL_MESSAGE in caplog.text) == in_caplog


@pytest.mark.parametrize(
    # Check that with or without hass behaves the same
    ("decorator", "in_caplog"),
    [
        (service.verify_domain_control, True),  # old pass-through
        (lambda _, domain: service.verify_domain_control(domain), False),  # new
    ],
)
async def test_domain_control_unauthorized(
    hass: HomeAssistant,
    hass_read_only_user: MockUser,
    decorator: Callable[[HomeAssistant, str], Any],
    in_caplog: bool,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test domain verification in a service call with an unauthorized user."""
    mock_registry(
        hass,
        {
            "light.kitchen": RegistryEntryWithDefaults(
                entity_id="light.kitchen",
                unique_id="kitchen",
                platform="test_domain",
            )
        },
    )

    calls = []

    async def mock_service_log(call):
        """Define a protected service."""
        calls.append(call)

    protected_mock_service = decorator(hass, "test_domain")(mock_service_log)

    hass.services.async_register(
        "test_domain", "test_service", protected_mock_service, schema=None
    )

    with pytest.raises(exceptions.Unauthorized):
        await hass.services.async_call(
            "test_domain",
            "test_service",
            {},
            blocking=True,
            context=Context(user_id=hass_read_only_user.id),
        )

    assert len(calls) == 0

    assert (_DEPRECATED_VERIFY_DOMAIN_CONTROL_MESSAGE in caplog.text) == in_caplog


@pytest.mark.parametrize(
    # Check that with or without hass behaves the same
    ("decorator", "in_caplog"),
    [
        (service.verify_domain_control, True),  # old pass-through
        (lambda _, domain: service.verify_domain_control(domain), False),  # new
    ],
)
async def test_domain_control_admin(
    hass: HomeAssistant,
    hass_admin_user: MockUser,
    decorator: Callable[[HomeAssistant, str], Any],
    in_caplog: bool,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test domain verification in a service call with an admin user."""
    mock_registry(
        hass,
        {
            "light.kitchen": RegistryEntryWithDefaults(
                entity_id="light.kitchen",
                unique_id="kitchen",
                platform="test_domain",
            )
        },
    )

    calls = []

    async def mock_service_log(call):
        """Define a protected service."""
        calls.append(call)

    protected_mock_service = decorator(hass, "test_domain")(mock_service_log)

    hass.services.async_register(
        "test_domain", "test_service", protected_mock_service, schema=None
    )

    await hass.services.async_call(
        "test_domain",
        "test_service",
        {},
        blocking=True,
        context=Context(user_id=hass_admin_user.id),
    )

    assert len(calls) == 1

    assert (_DEPRECATED_VERIFY_DOMAIN_CONTROL_MESSAGE in caplog.text) == in_caplog


@pytest.mark.parametrize(
    # Check that with or without hass behaves the same
    ("decorator", "in_caplog"),
    [
        (service.verify_domain_control, True),  # old pass-through
        (lambda _, domain: service.verify_domain_control(domain), False),  # new
    ],
)
async def test_domain_control_no_user(
    hass: HomeAssistant,
    decorator: Callable[[HomeAssistant, str], Any],
    in_caplog: bool,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """Test domain verification in a service call with no user."""
    mock_registry(
        hass,
        {
            "light.kitchen": RegistryEntryWithDefaults(
                entity_id="light.kitchen",
                unique_id="kitchen",
                platform="test_domain",
            )
        },
    )

    calls = []

    async def mock_service_log(call):
        """Define a protected service."""
        calls.append(call)

    protected_mock_service = decorator(hass, "test_domain")(mock_service_log)

    hass.services.async_register(
        "test_domain", "test_service", protected_mock_service, schema=None
    )

    await hass.services.async_call(
        "test_domain",
        "test_service",
        {},
        blocking=True,
        context=Context(user_id=None),
    )

    assert len(calls) == 1

    assert (_DEPRECATED_VERIFY_DOMAIN_CONTROL_MESSAGE in caplog.text) == in_caplog


async def test_extract_from_service_available_device(hass: HomeAssistant) -> None:
    """Test the extraction of entity from service and device is available."""
    entities = [
        MockEntity(name="test_1", entity_id="test_domain.test_1"),
        MockEntity(name="test_2", entity_id="test_domain.test_2", available=False),
        MockEntity(name="test_3", entity_id="test_domain.test_3"),
        MockEntity(name="test_4", entity_id="test_domain.test_4", available=False),
    ]

    call_1 = ServiceCall(hass, "test", "service", data={"entity_id": ENTITY_MATCH_ALL})

    assert [
        ent.entity_id
        for ent in (await service.async_extract_entities(hass, entities, call_1))
    ] == ["test_domain.test_1", "test_domain.test_3"]

    call_2 = ServiceCall(
        hass,
        "test",
        "service",
        data={"entity_id": ["test_domain.test_3", "test_domain.test_4"]},
    )

    assert [
        ent.entity_id
        for ent in (await service.async_extract_entities(hass, entities, call_2))
    ] == ["test_domain.test_3"]

    assert (
        await service.async_extract_entities(
            hass,
            entities,
            ServiceCall(
                hass,
                "test",
                "service",
                data={"entity_id": ENTITY_MATCH_NONE},
            ),
        )
        == []
    )


async def test_extract_from_service_empty_if_no_entity_id(hass: HomeAssistant) -> None:
    """Test the extraction from service without specifying entity."""
    entities = [
        MockEntity(name="test_1", entity_id="test_domain.test_1"),
        MockEntity(name="test_2", entity_id="test_domain.test_2"),
    ]
    call = ServiceCall(hass, "test", "service")

    assert [
        ent.entity_id
        for ent in (await service.async_extract_entities(hass, entities, call))
    ] == []


async def test_extract_from_service_filter_out_non_existing_entities(
    hass: HomeAssistant,
) -> None:
    """Test the extraction of non existing entities from service."""
    entities = [
        MockEntity(name="test_1", entity_id="test_domain.test_1"),
        MockEntity(name="test_2", entity_id="test_domain.test_2"),
    ]

    call = ServiceCall(
        hass,
        "test",
        "service",
        {"entity_id": ["test_domain.test_2", "test_domain.non_exist"]},
    )

    assert [
        ent.entity_id
        for ent in (await service.async_extract_entities(hass, entities, call))
    ] == ["test_domain.test_2"]


async def test_extract_from_service_area_id(
    hass: HomeAssistant, floor_area_mock
) -> None:
    """Test the extraction using area ID as reference."""
    entities = [
        MockEntity(name="in_area", entity_id="light.in_area"),
        MockEntity(name="no_area", entity_id="light.no_area"),
        MockEntity(name="diff_area", entity_id="light.diff_area"),
    ]

    call = ServiceCall(hass, "light", "turn_on", {"area_id": "test-area"})
    extracted = await service.async_extract_entities(hass, entities, call)
    assert len(extracted) == 1
    assert extracted[0].entity_id == "light.in_area"

    call = ServiceCall(
        hass, "light", "turn_on", {"area_id": ["test-area", "diff-area"]}
    )
    extracted = await service.async_extract_entities(hass, entities, call)
    assert len(extracted) == 2
    assert sorted(ent.entity_id for ent in extracted) == [
        "light.diff_area",
        "light.in_area",
    ]

    call = ServiceCall(
        hass,
        "light",
        "turn_on",
        {"area_id": ["test-area", "diff-area"], "device_id": "device-no-area-id"},
    )
    extracted = await service.async_extract_entities(hass, entities, call)
    assert len(extracted) == 3
    assert sorted(ent.entity_id for ent in extracted) == [
        "light.diff_area",
        "light.in_area",
        "light.no_area",
    ]


@pytest.mark.usefixtures("label_mock")
async def test_extract_from_service_label_id(hass: HomeAssistant) -> None:
    """Test the extraction using label ID as reference."""
    entities = [
        MockEntity(name="with_my_label", entity_id="light.with_my_label"),
        MockEntity(name="no_labels", entity_id="light.no_labels"),
        MockEntity(
            name="with_labels_from_device", entity_id="light.with_labels_from_device"
        ),
    ]

    call = ServiceCall(hass, "light", "turn_on", {"label_id": "label_area"})
    extracted = await service.async_extract_entities(hass, entities, call)
    assert len(extracted) == 1
    assert extracted[0].entity_id == "light.with_labels_from_device"

    call = ServiceCall(hass, "light", "turn_on", {"label_id": "my-label"})
    extracted = await service.async_extract_entities(hass, entities, call)
    assert len(extracted) == 1
    assert extracted[0].entity_id == "light.with_my_label"

    call = ServiceCall(hass, "light", "turn_on", {"label_id": ["my-label", "label1"]})
    extracted = await service.async_extract_entities(hass, entities, call)
    assert len(extracted) == 2
    assert sorted(ent.entity_id for ent in extracted) == [
        "light.with_labels_from_device",
        "light.with_my_label",
    ]

    call = ServiceCall(
        hass,
        "light",
        "turn_on",
        {"label_id": ["my-label", "label1"], "device_id": "device-no-labels"},
    )
    extracted = await service.async_extract_entities(hass, entities, call)
    assert len(extracted) == 3
    assert sorted(ent.entity_id for ent in extracted) == [
        "light.no_labels",
        "light.with_labels_from_device",
        "light.with_my_label",
    ]


async def test_entity_service_call_warn_referenced(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test we only warn for referenced entities in entity_service_call."""
    call = ServiceCall(
        hass,
        "light",
        "turn_on",
        {
            "area_id": "non-existent-area",
            "entity_id": "non.existent",
            "device_id": "non-existent-device",
            "floor_id": "non-existent-floor",
            "label_id": "non-existent-label",
        },
    )
    await service.entity_service_call(hass, {}, "", call)
    assert (
        "Referenced floors non-existent-floor, areas non-existent-area, "
        "devices non-existent-device, entities non.existent, "
        "labels non-existent-label are missing or not currently available"
    ) in caplog.text


async def test_async_extract_entities_warn_referenced(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test we only warn for referenced entities in async_extract_entities."""
    call = ServiceCall(
        hass,
        "light",
        "turn_on",
        {
            "area_id": "non-existent-area",
            "entity_id": "non.existent",
            "device_id": "non-existent-device",
            "floor_id": "non-existent-floor",
            "label_id": "non-existent-label",
        },
    )
    extracted = await service.async_extract_entities(hass, {}, call)
    assert len(extracted) == 0
    assert (
        "Referenced floors non-existent-floor, areas non-existent-area, "
        "devices non-existent-device, entities non.existent, "
        "labels non-existent-label are missing or not currently available"
    ) in caplog.text


async def test_async_extract_config_entry_ids(hass: HomeAssistant) -> None:
    """Test we can find devices that have no entities."""

    device_no_entities = dr.DeviceEntry(id="device-no-entities", config_entries={"abc"})

    call = ServiceCall(
        hass,
        "homeassistant",
        "reload_config_entry",
        {
            "device_id": "device-no-entities",
        },
    )

    mock_device_registry(
        hass,
        {
            device_no_entities.id: device_no_entities,
        },
    )

    assert await service.async_extract_config_entry_ids(hass, call) == {"abc"}


async def test_reload_service_helper(hass: HomeAssistant) -> None:
    """Test the reload service helper."""

    active_reload_calls = 0
    reloaded = []

    async def reload_service_handler(service_call: ServiceCall) -> None:
        """Remove all automations and load new ones from config."""
        nonlocal active_reload_calls
        # Assert the reload helper prevents parallel reloads
        assert not active_reload_calls
        active_reload_calls += 1
        if not (target := service_call.data.get("target")):
            reloaded.append("all")
        else:
            reloaded.append(target)
        await asyncio.sleep(0.01)
        active_reload_calls -= 1

    def reload_targets(service_call: ServiceCall) -> set[str | None]:
        if target_id := service_call.data.get("target"):
            return {target_id}
        return {"target1", "target2", "target3", "target4"}

    # Test redundant reload of single targets
    reloader = service.ReloadServiceHelper(reload_service_handler, reload_targets)
    tasks = [
        # This reload task will start executing first, (target1)
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        # These reload tasks will be deduplicated to (target2, target3, target4, target1)
        # while the first task is reloaded, note that target1 can't be deduplicated
        # because it's already being reloaded.
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
    ]
    await asyncio.gather(*tasks)
    assert reloaded == unordered(
        ["target1", "target2", "target3", "target4", "target1"]
    )

    # Test redundant reload of multiple targets + single target
    reloaded.clear()
    tasks = [
        # This reload task will start executing first, (target1)
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        # These reload tasks will be deduplicated to (target2, target3, target4, all)
        # while the first task is reloaded.
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
        reloader.execute_service(ServiceCall(hass, "test", "test")),
    ]
    await asyncio.gather(*tasks)
    assert reloaded == unordered(["target1", "target2", "target3", "target4", "all"])

    # Test redundant reload of multiple targets + single target
    reloaded.clear()
    tasks = [
        # This reload task will start executing first, (all)
        reloader.execute_service(ServiceCall(hass, "test", "test")),
        # These reload tasks will be deduplicated to (target1, target2, target3, target4)
        # while the first task is reloaded.
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
    ]
    await asyncio.gather(*tasks)
    assert reloaded == unordered(["all", "target1", "target2", "target3", "target4"])

    # Test redundant reload of single targets
    reloaded.clear()
    tasks = [
        # This reload task will start executing first, (target1)
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        # These reload tasks will be deduplicated to (target2, target3, target4, target1)
        # while the first task is reloaded, note that target1 can't be deduplicated
        # because it's already being reloaded.
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
    ]
    await asyncio.gather(*tasks)
    assert reloaded == unordered(
        ["target1", "target2", "target3", "target4", "target1"]
    )

    # Test redundant reload of multiple targets + single target
    reloaded.clear()
    tasks = [
        # This reload task will start executing first, (target1)
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        # These reload tasks will be deduplicated to (target2, target3, target4, all)
        # while the first task is reloaded.
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
        reloader.execute_service(ServiceCall(hass, "test", "test")),
        reloader.execute_service(ServiceCall(hass, "test", "test")),
    ]
    await asyncio.gather(*tasks)
    assert reloaded == unordered(["target1", "target2", "target3", "target4", "all"])

    # Test redundant reload of multiple targets + single target
    reloaded.clear()
    tasks = [
        # This reload task will start executing first, (all)
        reloader.execute_service(ServiceCall(hass, "test", "test")),
        # These reload tasks will be deduplicated to (target1, target2, target3, target4)
        # while the first task is reloaded.
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target1"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target2"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target3"})
        ),
        reloader.execute_service(
            ServiceCall(hass, "test", "test", {"target": "target4"})
        ),
    ]
    await asyncio.gather(*tasks)
    assert reloaded == unordered(["all", "target1", "target2", "target3", "target4"])


async def test_deprecated_service_target_selector_class(hass: HomeAssistant) -> None:
    """Test that the deprecated ServiceTargetSelector class forwards correctly."""
    call = ServiceCall(
        hass,
        "test",
        "test",
        {
            "entity_id": ["light.test", "switch.test"],
            "area_id": "kitchen",
            "device_id": ["device1", "device2"],
            "floor_id": "first_floor",
            "label_id": ["label1", "label2"],
        },
    )
    selector = service.ServiceTargetSelector(call)

    assert selector.entity_ids == {"light.test", "switch.test"}
    assert selector.area_ids == {"kitchen"}
    assert selector.device_ids == {"device1", "device2"}
    assert selector.floor_ids == {"first_floor"}
    assert selector.label_ids == {"label1", "label2"}
    assert selector.has_any_selector is True


async def test_deprecated_selected_entities_class(
    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None:
    """Test that the deprecated SelectedEntities class forwards correctly."""
    selected = service.SelectedEntities(
        referenced={"entity.test"},
        indirectly_referenced=set(),
        referenced_devices=set(),
        referenced_areas=set(),
        missing_devices={"missing_device"},
        missing_areas={"missing_area"},
        missing_floors={"missing_floor"},
        missing_labels={"missing_label"},
    )

    missing_entities = {"entity.missing"}
    selected.log_missing(missing_entities)
    assert (
        "Referenced floors missing_floor, areas missing_area, "
        "devices missing_device, entities entity.missing, "
        "labels missing_label are missing or not currently available" in caplog.text
    )


async def test_deprecated_async_extract_referenced_entity_ids(
    hass: HomeAssistant,
) -> None:
    """Test that the deprecated async_extract_referenced_entity_ids function forwards correctly."""
    from homeassistant.helpers import target  # noqa: PLC0415

    mock_selected = target.SelectedEntities(
        referenced={"entity.test"},
        indirectly_referenced={"entity.indirect"},
    )
    with patch(
        "homeassistant.helpers.target.async_extract_referenced_entity_ids",
        return_value=mock_selected,
    ) as mock_target_func:
        call = ServiceCall(hass, "test", "test", {"entity_id": "light.test"})
        result = service.async_extract_referenced_entity_ids(
            hass, call, expand_group=False
        )

        # Verify target helper was called with correct parameters
        mock_target_func.assert_called_once()
        args = mock_target_func.call_args
        assert args[0][0] is hass
        assert args[0][1].entity_ids == {"light.test"}
        assert args[0][2] is False

        assert dataclasses.asdict(result) == dataclasses.asdict(mock_selected)


async def test_register_platform_entity_service(
    hass: HomeAssistant,
) -> None:
    """Test registering a platform entity service."""
    entities = []

    @callback
    def handle_service(entity, *_):
        entities.append(entity)

    service.async_register_platform_entity_service(
        hass,
        "mock_platform",
        "hello",
        entity_domain="mock_integration",
        schema={},
        func=handle_service,
        description_placeholders={"test_placeholder": "beer"},
    )
    descriptions = await service.async_get_all_descriptions(hass)
    assert (
        descriptions["mock_platform"]["hello"]["description_placeholders"]
        == {"test_placeholder": "beer"}
        == {"test_placeholder": "beer"}
    )

    await hass.services.async_call(
        "mock_platform", "hello", {"entity_id": "all"}, blocking=True
    )
    assert entities == []

    entity_platform = MockEntityPlatform(
        hass, domain="mock_integration", platform_name="mock_platform", platform=None
    )
    entity1 = MockEntity(entity_id="mock_integration.entity1")
    entity2 = MockEntity(entity_id="mock_integration.entity2")
    await entity_platform.async_add_entities([entity1, entity2])

    await hass.services.async_call(
        "mock_platform", "hello", {"entity_id": "all"}, blocking=True
    )

    assert entities == unordered([entity1, entity2])


async def test_register_platform_entity_service_response_data(
    hass: HomeAssistant,
) -> None:
    """Test an entity service that supports response data."""

    async def generate_response(
        target: MockEntity, call: ServiceCall
    ) -> ServiceResponse:
        assert call.return_response
        return {"response-key": "response-value"}

    service.async_register_platform_entity_service(
        hass,
        "mock_platform",
        "hello",
        entity_domain="mock_integration",
        schema={"some": str},
        func=generate_response,
        supports_response=SupportsResponse.ONLY,
    )

    entity_platform = MockEntityPlatform(
        hass, domain="mock_integration", platform_name="mock_platform", platform=None
    )
    entity = MockEntity(entity_id="mock_integration.entity")
    await entity_platform.async_add_entities([entity])

    response_data = await hass.services.async_call(
        "mock_platform",
        "hello",
        service_data={"some": "data"},
        target={"entity_id": [entity.entity_id]},
        blocking=True,
        return_response=True,
    )
    assert response_data == {
        "mock_integration.entity": {"response-key": "response-value"}
    }


async def test_register_platform_entity_service_response_data_multiple_matches(
    hass: HomeAssistant,
) -> None:
    """Test an entity service with response data and matching many entities."""

    async def generate_response(
        target: MockEntity, call: ServiceCall
    ) -> ServiceResponse:
        assert call.return_response
        return {"response-key": f"response-value-{target.entity_id}"}

    service.async_register_platform_entity_service(
        hass,
        "mock_platform",
        "hello",
        entity_domain="mock_integration",
        schema={"some": str},
        func=generate_response,
        supports_response=SupportsResponse.ONLY,
    )

    entity_platform = MockEntityPlatform(
        hass, domain="mock_integration", platform_name="mock_platform", platform=None
    )
    entity1 = MockEntity(entity_id="mock_integration.entity1")
    entity2 = MockEntity(entity_id="mock_integration.entity2")
    await entity_platform.async_add_entities([entity1, entity2])

    response_data = await hass.services.async_call(
        "mock_platform",
        "hello",
        service_data={"some": "data"},
        target={"entity_id": [entity1.entity_id, entity2.entity_id]},
        blocking=True,
        return_response=True,
    )
    assert response_data == {
        "mock_integration.entity1": {
            "response-key": "response-value-mock_integration.entity1"
        },
        "mock_integration.entity2": {
            "response-key": "response-value-mock_integration.entity2"
        },
    }


async def test_register_platform_entity_service_response_data_multiple_matches_raises(
    hass: HomeAssistant,
) -> None:
    """Test entity service response matching many entities raises."""

    async def generate_response(
        target: MockEntity, call: ServiceCall
    ) -> ServiceResponse:
        assert call.return_response
        if target.entity_id == "mock_integration.entity1":
            raise RuntimeError("Something went wrong")
        return {"response-key": f"response-value-{target.entity_id}"}

    service.async_register_platform_entity_service(
        hass,
        "mock_platform",
        "hello",
        entity_domain="mock_integration",
        schema={"some": str},
        func=generate_response,
        supports_response=SupportsResponse.ONLY,
    )

    entity_platform = MockEntityPlatform(
        hass, domain="mock_integration", platform_name="mock_platform", platform=None
    )
    entity1 = MockEntity(entity_id="mock_integration.entity1")
    entity2 = MockEntity(entity_id="mock_integration.entity2")
    await entity_platform.async_add_entities([entity1, entity2])

    with pytest.raises(RuntimeError, match="Something went wrong"):
        await hass.services.async_call(
            "mock_platform",
            "hello",
            service_data={"some": "data"},
            target={"entity_id": [entity1.entity_id, entity2.entity_id]},
            blocking=True,
            return_response=True,
        )


async def test_register_platform_entity_service_limited_to_matching_platforms(
    hass: HomeAssistant,
    entity_registry: er.EntityRegistry,
    area_registry: ar.AreaRegistry,
) -> None:
    """Test entity services only target entities for the platform and domain."""

    mock_area = area_registry.async_get_or_create("mock_area")

    entity1_entry = entity_registry.async_get_or_create(
        "base_platform", "mock_platform", "1234", suggested_object_id="entity1"
    )
    entity_registry.async_update_entity(entity1_entry.entity_id, area_id=mock_area.id)
    entity2_entry = entity_registry.async_get_or_create(
        "base_platform", "mock_platform", "5678", suggested_object_id="entity2"
    )
    entity_registry.async_update_entity(entity2_entry.entity_id, area_id=mock_area.id)
    entity3_entry = entity_registry.async_get_or_create(
        "base_platform", "other_mock_platform", "7891", suggested_object_id="entity3"
    )
    entity_registry.async_update_entity(entity3_entry.entity_id, area_id=mock_area.id)
    entity4_entry = entity_registry.async_get_or_create(
        "base_platform", "other_mock_platform", "1433", suggested_object_id="entity4"
    )
    entity_registry.async_update_entity(entity4_entry.entity_id, area_id=mock_area.id)

    async def generate_response(
        target: MockEntity, call: ServiceCall
    ) -> ServiceResponse:
        assert call.return_response
        return {"response-key": f"response-value-{target.entity_id}"}

    service.async_register_platform_entity_service(
        hass,
        "mock_platform",
        "hello",
        entity_domain="base_platform",
        schema={"some": str},
        func=generate_response,
        supports_response=SupportsResponse.ONLY,
    )

    entity_platform = MockEntityPlatform(
        hass, domain="base_platform", platform_name="mock_platform", platform=None
    )
    entity1 = MockEntity(
        entity_id=entity1_entry.entity_id, unique_id=entity1_entry.unique_id
    )
    entity2 = MockEntity(
        entity_id=entity2_entry.entity_id, unique_id=entity2_entry.unique_id
    )
    await entity_platform.async_add_entities([entity1, entity2])

    other_entity_platform = MockEntityPlatform(
        hass, domain="base_platform", platform_name="other_mock_platform", platform=None
    )
    entity3 = MockEntity(
        entity_id=entity3_entry.entity_id, unique_id=entity3_entry.unique_id
    )
    entity4 = MockEntity(
        entity_id=entity4_entry.entity_id, unique_id=entity4_entry.unique_id
    )
    await other_entity_platform.async_add_entities([entity3, entity4])

    response_data = await hass.services.async_call(
        "mock_platform",
        "hello",
        service_data={"some": "data"},
        target={"area_id": [mock_area.id]},
        blocking=True,
        return_response=True,
    )
    # We should not target entity3 and entity4 even though they are in the area
    # because they are only part of the domain and not the platform
    assert response_data == {
        "base_platform.entity1": {
            "response-key": "response-value-base_platform.entity1"
        },
        "base_platform.entity2": {
            "response-key": "response-value-base_platform.entity2"
        },
    }


async def test_register_platform_entity_service_none_schema(
    hass: HomeAssistant,
) -> None:
    """Test registering a service with schema set to None."""
    entities = []

    @callback
    def handle_service(entity, *_):
        entities.append(entity)

    service.async_register_platform_entity_service(
        hass,
        "mock_platform",
        "hello",
        entity_domain="mock_integration",
        schema=None,
        func=handle_service,
    )

    entity_platform = MockEntityPlatform(
        hass, domain="mock_integration", platform_name="mock_platform", platform=None
    )
    entity1 = MockEntity(name="entity_1")
    entity2 = MockEntity(name="entity_1")
    await entity_platform.async_add_entities([entity1, entity2])

    await hass.services.async_call(
        "mock_platform", "hello", {"entity_id": "all"}, blocking=True
    )

    assert len(entities) == 2
    assert entity1 in entities
    assert entity2 in entities


async def test_register_platform_entity_service_non_entity_service_schema(
    hass: HomeAssistant,
) -> None:
    """Test attempting to register a service with a non entity service schema."""
    expected_message = "registers an entity service with a non entity service schema"

    for idx, schema in enumerate(
        (
            vol.Schema({"some": str}),
            vol.All(vol.Schema({"some": str})),
            vol.Any(vol.Schema({"some": str})),
        )
    ):
        with pytest.raises(HomeAssistantError, match=expected_message):
            service.async_register_platform_entity_service(
                hass,
                "mock_platform",
                f"hello_{idx}",
                entity_domain="mock_integration",
                schema=schema,
                func=Mock(),
            )

    for idx, schema in enumerate(
        (
            cv.make_entity_service_schema({"some": str}),
            vol.Schema(cv.make_entity_service_schema({"some": str})),
            vol.All(cv.make_entity_service_schema({"some": str})),
        )
    ):
        service.async_register_platform_entity_service(
            hass,
            "mock_platform",
            f"test_service_{idx}",
            entity_domain="mock_integration",
            schema=schema,
            func=Mock(),
        )
